package net.zkbc.helloworld;

import static net.zkbc.akka.SpringExtension.SpringExtProvider;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.cluster.sharding.ClusterSharding;
import akka.cluster.sharding.ClusterShardingSettings;
import akka.cluster.sharding.ShardRegion;
import net.zkbc.helloworld.actor.HitsActor;

@Configuration
public class AppConfiguration {

	@Bean
	@Autowired
	public ActorRef hitsActorShardRegion(ActorSystem actorSystem) {
		ShardRegion.MessageExtractor messageExtractor = new ShardRegion.MessageExtractor() {
			@Override
			public Object entityMessage(Object message) {
				return message;
			}

			@Override
			public String entityId(Object message) {
				if (message instanceof HitsActor.Count) {
					return ((HitsActor.Count) message).getFrom();
				}
				return null;
			}

			@Override
			public String shardId(Object message) {
				if (message instanceof HitsActor.Count) {
					return "shard-" + ((HitsActor.Count) message).getFrom().substring(0, 1);
				}
				return null;
			}

		};

		return startShardRegion(actorSystem, "hitsActor", messageExtractor);
	}

	private ActorRef startShardRegion(ActorSystem actorSystem, String typeName,
			ShardRegion.MessageExtractor messageExtractor) {
		return ClusterSharding.get(actorSystem).start(typeName, SpringExtProvider.get(actorSystem).props(typeName),
				ClusterShardingSettings.create(actorSystem), messageExtractor);
	}

}
